package com.zhang.hadoop.flink.test7;

import com.zhang.hadoop.flink.base.ClickSource;
import com.zhang.hadoop.flink.base.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * @author: zhang yufei
 * @createTime:2022/7/10 15:06
 * @description:
 */
public class SplitStreamTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        })
                );

        //定义输出标签
        OutputTag<Tuple3<String, String, Long>> yanghuiTag = new OutputTag<Tuple3<String, String, Long>>("yanghui") {};
        OutputTag<Tuple3<String, String, Long>> yupingTag = new OutputTag<Tuple3<String, String, Long>>("yuping") {};

        SingleOutputStreamOperator<Event> processedStream = stream.process(new ProcessFunction<Event, Event>() {
            @Override
            public void processElement(Event event, Context context, Collector<Event> collector) throws Exception {
                if (event.user.equals("欧阳慧")) {
                    context.output(yanghuiTag, Tuple3.of(event.user, event.url, event.timestamp));
                } else if (event.user.equals("陈玉萍")) {
                    context.output(yupingTag, Tuple3.of(event.user, event.url, event.timestamp));
                } else {
                    collector.collect(event);
                }
            }
        });

        processedStream.print("else");
        processedStream.getSideOutput(yanghuiTag).print("ouyuang");
        processedStream.getSideOutput(yupingTag).print("yuping");

        env.execute();
    }
}
